package cn.itcast.es.utils;

import cn.itcast.es.annotations.ID;
import cn.itcast.es.annotations.Index;
import cn.itcast.es.pojo.PageDTO;
import com.alibaba.fastjson.JSON;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.completion.CompletionSuggestion;
import org.elasticsearch.search.suggest.completion.CompletionSuggestionBuilder;
import org.springframework.util.CollectionUtils;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @author 虎哥
 */
public class ElasticsearchUtil<T> {

    private RestHighLevelClient restHighLevelClient;
    private Class<T> clazz;
    private Field idField;
    private String indexName;

    public ElasticsearchUtil(RestHighLevelClient restHighLevelClient) {
        this.restHighLevelClient = restHighLevelClient;
        // 获取父类的泛型类型
        ParameterizedType genericSuperclass = (ParameterizedType) this.getClass().getGenericSuperclass();
        Type[] types = genericSuperclass.getActualTypeArguments();
        clazz = (Class<T>) types[0];

        // 获取索引库名称
        if(clazz.isAnnotationPresent(Index.class)){
            // 获取注解
            Index index = clazz.getAnnotation(Index.class);
            indexName = index.value();
        }

        // 获取ID字段
        Field[] declaredFields = clazz.getDeclaredFields();
        for (Field field : declaredFields) {
            // 判断哪个是id
            field.setAccessible(true);
            if (field.isAnnotationPresent(ID.class)) {
                idField = field;
            }
        }
    }

    public void createIndex(String source) {
        // 准备请求对象
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        // 指定source
        request.source(source, XContentType.JSON);
        // 发请求
        try {
            restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void deleteIndex() {
        try {
            // 发请求
            restHighLevelClient.indices().delete(new DeleteIndexRequest(indexName), RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public <T> void save(T goods) {
        // 准备请求对象
        IndexRequest request = new IndexRequest(indexName)
                .id(getId(goods))
                .source(JSON.toJSONString(goods), XContentType.JSON);
        try {
            // 发出请求
            restHighLevelClient.index(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private String getId(Object goods) {
        try {
            Object id = idField.get(goods);
            if (id == null) {
                throw new RuntimeException("ID不能为空");
            }
            return id.toString();
        } catch (IllegalAccessException e) {
            throw new RuntimeException(e);
        }
    }

    public void saveBatch(List<T> goodsList) {
        // 准备请求对象
        BulkRequest request = new BulkRequest(indexName);
        for (T t : goodsList) {
            request.add(new IndexRequest()
                    .id(getId(t).toString())
                    .source(JSON.toJSONString(t), XContentType.JSON));
        }
        try {
            // 发出请求
            restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void deleteById(String id) {
        // 准备请求对象
        DeleteRequest request = new DeleteRequest(indexName, id);
        try {
            // 发出请求
            restHighLevelClient.delete(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public Mono<T> getByIdAsync(String id) {
        // 准备请求对象
        GetRequest request = new GetRequest(indexName, id);
        // 用Mono封装调用的逻辑
        return Mono.create(monoSink -> {
            // 发出请求
            restHighLevelClient.getAsync(request, RequestOptions.DEFAULT, new ActionListener<GetResponse>() {
                @Override
                public void onResponse(GetResponse response) {
                    // 解析响应
                    String json = response.getSourceAsString();
                    // 反序列化
                    T t = JSON.parseObject(json, clazz);
                    // 发射数据
                    monoSink.success(t);
                }

                @Override
                public void onFailure(Exception e) {
                    monoSink.error(e);
                }
            });
        });
    }

    public Mono<PageDTO<T>> searchAsync(SearchSourceBuilder sourceBuilder) {
        // 准备请求对象
        SearchRequest request = new SearchRequest(indexName);
        request.source(sourceBuilder);

        return Mono.create(monoSink -> {
            // 发请求
            restHighLevelClient.searchAsync(request, RequestOptions.DEFAULT, new ActionListener<SearchResponse>() {
                @Override
                public void onResponse(SearchResponse response) {
                    // 解析响应
                    PageDTO<T> pageDTO = parseSearchResponse(response);
                    // 发射数据
                    monoSink.success(pageDTO);
                }

                @Override
                public void onFailure(Exception e) {
                    monoSink.error(e);
                }
            });
        });
    }

    public Mono<List<String>> suggest(String suggestField, String prefix) {
        // 准备请求对象
        SearchRequest request = new SearchRequest(indexName);
        // 准备请求参数
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
                .fetchSource(new String[0], null)
                .suggest(new SuggestBuilder().addSuggestion(
                        "mySuggest",
                        new CompletionSuggestionBuilder(suggestField).size(20).skipDuplicates(true).prefix(prefix)
                ));
        request.source(sourceBuilder);

        return Mono.create(monoSink -> {
            // 发出请求
            restHighLevelClient.searchAsync(request, RequestOptions.DEFAULT, new ActionListener<SearchResponse>() {
                @Override
                public void onResponse(SearchResponse response) {
                    // 解析结果
                    List<String> list = parseSuggestResponse(response);
                    // 发射
                    monoSink.success(list);
                }

                @Override
                public void onFailure(Exception e) {
                    monoSink.error(e);
                }
            });
        });
    }

    private List<String> parseSuggestResponse(SearchResponse response) {
        Suggest suggest = response.getSuggest();
        CompletionSuggestion suggestion = suggest.getSuggestion("mySuggest");
        List<String> list = new ArrayList<>(suggestion.getOptions().size());
        for (CompletionSuggestion.Entry.Option option : suggestion.getOptions()) {
            list.add(option.getText().toString());
        }
        return list;
    }

    private PageDTO<T> parseSearchResponse(SearchResponse response) {
        // 解析结果
        SearchHits searchHits = response.getHits();

        // 解析总条数
        long total = searchHits.getTotalHits().value;

        // 解析当前页数据
        SearchHit[] hits = searchHits.getHits();
        List<T> goodsList = new ArrayList<>(hits.length);

        // 处理hit，转为goods
        for (SearchHit hit : hits) {
            String json = hit.getSourceAsString();
            // 反序列化
            T t = JSON.parseObject(json, clazz);
            // 添加
            goodsList.add(t);
            // 处理高亮
            Map<String, HighlightField> highlightFields = hit.getHighlightFields();
            if (!CollectionUtils.isEmpty(highlightFields)) {
                for (HighlightField highlightField : highlightFields.values()) {
                    String fieldName = highlightField.getName();
                    String fieldValue = StringUtils.join(highlightField.getFragments());
                    // 设置高亮字段
                    try {
                        BeanUtils.setProperty(t, fieldName, fieldValue);
                    } catch (Exception e) {
                        throw new RuntimeException("高亮处理失败");
                    }
                }
            }
        }

        // 返回值
        return new PageDTO<>(total, goodsList);
    }
}
